package cn.itcast.job;

import cn.itcast.avro.AvroDeserializationSchema;
import cn.itcast.avro.SseAvro;
import cn.itcast.avro.SzseAvro;
import cn.itcast.bean.CleanBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.map.SseMap;
import cn.itcast.map.SzseMap;
import cn.itcast.task.*;
import cn.itcast.util.QuotUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;

import java.math.BigDecimal;
import java.util.Properties;


//个股业务开发：秒级、分时、K线、分时数据备份、个股涨跌幅行情
public class StockStream {
    /**
     * 开发步骤：
     *   1.创建StockStream单例对象，创建main方法
     *   2.获取流处理执行环境
     *   3.设置事件时间
     *   4.设置检查点机制
     *   5.设置重启机制
     *   6.触发执行
     */

    //todo 1.创建StockStream单例对象，创建main方法
    public static void main(String[] args) throws Exception {
        //todo 2.获取流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //todo 3.设置事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);//便于开发和测试

        //todo 4.设置检查点机制(只用于生产环境）
        //env.enableCheckpointing(50001);//开启检查点,设置检查点制作间隔时间
        //env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//强一致性
        //env.getCheckpointConfig().setFailOnCheckpointingErrors(false);//检查点制作失败，任务继续运行
        //env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);//最大线程数
        ////任务取消的时候，保留检查点，需要手动删除老的检查点
        //env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        ////设置检查点存储
        //env.setStateBackend(new FsStateBackend("hdfs://node01:8020/checkpoint/stock"));


        //todo 5.设置重启机制.
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(5)));


        //todo 6.整合kafka
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers",QuotConfig.config.getProperty("bootstrap.servers"));
        prop.setProperty("zookeeper.connect",QuotConfig.config.getProperty("zookeeper.connect"));

        FlinkKafkaConsumer011<SseAvro> sseKafkaConsumer = new FlinkKafkaConsumer011<SseAvro>(
                QuotConfig.config.getProperty("sse.topic"),
                new AvroDeserializationSchema(QuotConfig.config.getProperty("sse.topic")),
                prop);
        FlinkKafkaConsumer011<SzseAvro> szseKafkaConsumer = new FlinkKafkaConsumer011<SzseAvro>(
                QuotConfig.config.getProperty("szse.topic"),
                new AvroDeserializationSchema(QuotConfig.config.getProperty("szse.topic")),
                prop);

        sseKafkaConsumer.setStartFromEarliest();
        //szseKafkaConsumer.setStartFromEarliest();

        //加载数据
        DataStreamSource<SseAvro> sseSource = env.addSource(sseKafkaConsumer);
        DataStreamSource<SzseAvro> szseSource = env.addSource(szseKafkaConsumer);


        //todo 7.数据过滤
        SingleOutputStreamOperator<SseAvro> sseFilterData = sseSource.filter(new FilterFunction<SseAvro>() {
            @Override
            public boolean filter(SseAvro sseAvro) throws Exception {
                return QuotUtil.checkTime(sseAvro) && QuotUtil.checkData(sseAvro);
            }
        });

        SingleOutputStreamOperator<SzseAvro> szseFilterData = szseSource.filter(new FilterFunction<SzseAvro>() {
            @Override
            public boolean filter(SzseAvro szseAvro) throws Exception {
                return QuotUtil.checkTime(szseAvro) && QuotUtil.checkData(szseAvro);
            }
        });


        //todo 8.数据合并，使用union数据类型要一致
        DataStream<CleanBean> unionData = sseFilterData.map(new SseMap()).union(szseFilterData.map(new SzseMap()));


        //todo 9.过滤个股数据
        SingleOutputStreamOperator<CleanBean> filterData = unionData.filter(new FilterFunction<CleanBean>() {
            @Override
            public boolean filter(CleanBean cleanBean) throws Exception {
                return QuotUtil.isStock(cleanBean);
            }
        });

        //todo 10.设置水位线
        DataStream<CleanBean> waterData = filterData.assignTimestampsAndWatermarks(
                new BoundedOutOfOrdernessTimestampExtractor<CleanBean>(org.apache.flink.streaming.api.windowing.time.Time.seconds(Integer.parseInt(QuotConfig.config.getProperty("delay.time")))) {
                    @Override
                    public long extractTimestamp(CleanBean cleanBean) {
                        if(cleanBean.getSecName().equals("华泰股份")){
                            String secName = cleanBean.getSecName();
                            BigDecimal preClosePrice = cleanBean.getPreClosePrice();
                            BigDecimal openPrice = cleanBean.getOpenPrice();
                            BigDecimal tradePrice = cleanBean.getTradePrice();
                            BigDecimal maxPrice = cleanBean.getMaxPrice();
                            BigDecimal minPrice = cleanBean.getMinPrice();
                            Long vol = cleanBean.getTradeVolumn();
                            Long amt = cleanBean.getTradeAmt();
                            System.out.println(secName+" 前收盘价："+preClosePrice+"  开盘价:"+openPrice+"  现价:"+tradePrice+"  最高价:"+maxPrice+"  最低价:"+minPrice+"  成交量:"+vol+"  成交金额:"+amt);
                        }
                        return cleanBean.getEventTime();
                    }
                });

        //waterData.print();
        /**
         * 1.秒级行情(5s)
         * 2.分时行情(60s)
         * 3.分时行情备份
         * 4.个股涨跌幅
         * 5.个股K线
         */
        //new StockSecTask().process(waterData);
        new StockMinTask().process(waterData);
        //new StockMinHdfsTask().process(waterData); //备份
        new StockIncrTask().process(waterData);//涨跌幅
        //new StockKlineTask().process(waterData);

        //todo 6.触发执行
        env.execute("stock stream");
    }
}
